use std::net::SocketAddr;
use std::sync::Arc;

use bytes::Bytes;
use http_body_util::{BodyExt, Full};
use hyper::body::Incoming;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Method, Request, Response, StatusCode};
use hyper_util::rt::tokio::TokioIo;
use std::future::Future;
use tokio::net::TcpListener;
use tracing::{info, warn};

use pdata::internal::data::protogen::collector::logs::v1::{
    ExportLogsServiceRequest, ExportLogsServiceResponse,
};
use pdata::internal::data::protogen::collector::metrics::v1::{
    ExportMetricsServiceRequest, ExportMetricsServiceResponse,
};
use pdata::internal::data::protogen::collector::trace::v1::{
    ExportTraceServiceRequest, ExportTraceServiceResponse,
};
use prost::Message;

use crate::consumer::{ConsumeMetrics, ConsumeTraces, ConsumeLogs};

/// OTLP HTTP receiver that accepts telemetry data over HTTP protocol.
pub struct OtlpHttpReceiver {
    traces_consumer: Option<Arc<dyn ConsumeTraces + Send + Sync + 'static>>,
    metrics_consumer: Option<Arc<dyn ConsumeMetrics + Send + Sync + 'static>>,
    logs_consumer: Option<Arc<dyn ConsumeLogs + Send + Sync + 'static>>,
}

impl OtlpHttpReceiver {
    /// Create a new OTLP HTTP receiver with the specified consumers.
    pub fn new(
        traces_consumer: Option<Arc<dyn ConsumeTraces + Send + Sync + 'static>>,
        metrics_consumer: Option<Arc<dyn ConsumeMetrics + Send + Sync + 'static>>,
        logs_consumer: Option<Arc<dyn ConsumeLogs + Send + Sync + 'static>>,
    ) -> Self {
        Self {
            traces_consumer,
            metrics_consumer,
            logs_consumer,
        }
    }

    /// Start the HTTP receiver server on the specified address.
    pub async fn run(
        &self,
        addr: SocketAddr,
    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        info!("🚀 Starting OTLP HTTP Receiver on http://{}", addr);

        let listener = TcpListener::bind(addr).await?;
        let traces_consumer = self.traces_consumer.clone();
        let metrics_consumer = self.metrics_consumer.clone();
        let logs_consumer = self.logs_consumer.clone();

        loop {
            let (stream, _) = listener.accept().await?;
            let io = TokioIo::new(stream);
            let traces_consumer = traces_consumer.clone();
            let metrics_consumer = metrics_consumer.clone();
            let logs_consumer = logs_consumer.clone();

            tokio::spawn(async move {
                let service = service_fn(move |req: Request<Incoming>| {
                    let traces_consumer = traces_consumer.clone();
                    let metrics_consumer = metrics_consumer.clone();
                    let logs_consumer = logs_consumer.clone();
                    async move {
                        Ok::<_, std::convert::Infallible>(
                            handle_request(req, traces_consumer, metrics_consumer, logs_consumer).await,
                        )
                    }
                });
                let _ = http1::Builder::new().serve_connection(io, service).await;
            });
        }
    }
}

async fn handle_request(
    req: Request<Incoming>,
    traces_consumer: Option<Arc<dyn ConsumeTraces + Send + Sync + 'static>>,
    metrics_consumer: Option<Arc<dyn ConsumeMetrics + Send + Sync + 'static>>,
    logs_consumer: Option<Arc<dyn ConsumeLogs + Send + Sync + 'static>>,
) -> Response<Full<Bytes>> {
    // info!("🔗 HTTP connection from: {:?}", req);

    // Only accept POST
    if req.method() != Method::POST {
        return Response::builder()
            .status(StatusCode::METHOD_NOT_ALLOWED)
            .body(Full::from(Bytes::from_static(b"method not allowed")))
            .unwrap();
    }

    let path = req.uri().path();
    match path {
        "/v1/traces" => {
            let consume = traces_consumer.as_ref().map(|c| {
                let c = Arc::clone(c);
                move |data:ExportTraceServiceRequest| async move {
                    c.consume(data).await;
                }
            });
            handle_post_signal::<ExportTraceServiceRequest, ExportTraceServiceResponse, _, _>(
                req, consume,
            )
            .await
        }
        "/v1/metrics" => {
            let consume = metrics_consumer.as_ref().map(|c| {
                let c = Arc::clone(c);
                move |data: ExportMetricsServiceRequest| async move {
                    c.consume(data).await;
                }
            });
            handle_post_signal::<ExportMetricsServiceRequest, ExportMetricsServiceResponse, _, _>(
                req, consume,
            )
            .await
        }
        "/v1/logs" => {
            let consume = logs_consumer.as_ref().map(|c| {
                let c = Arc::clone(c);
                move |data: ExportLogsServiceRequest| async move {
                    c.consume(data).await;
                }
            });
            handle_post_signal::<ExportLogsServiceRequest, ExportLogsServiceResponse, _, _>(req, consume).await
        }
        _ => Response::builder()
            .status(StatusCode::NOT_FOUND)
            .body(Full::from(Bytes::from_static(b"not found")))
            .unwrap(),
    }
}

async fn handle_post_signal<MReq, MResp, F, Fut>(
    req: Request<Incoming>,
    consume: Option<F>,
) -> Response<Full<Bytes>>
where
    MReq: Message + Default + 'static,
    MResp: Message + Default,
    F: FnOnce(MReq) -> Fut,
    Fut: Future<Output = ()>,
{
    // We need the headers later, so split the request into parts and body first
    let (parts, body) = req.into_parts();
    let Ok(collected) = body.collect().await else {
        return Response::builder()
            .status(StatusCode::BAD_REQUEST)
            .body(Full::from(Bytes::from_static(b"invalid body")))
            .unwrap();
    };
    let bytes = collected.to_bytes();

    // If request is gzip-compressed, decompress before decoding protobuf
    let is_gzip = parts
        .headers
        .get(hyper::header::CONTENT_ENCODING)
        .and_then(|v| v.to_str().ok())
        .map(|v| v.eq_ignore_ascii_case("gzip"))
        .unwrap_or(false);

    let body_bytes: Vec<u8> = if is_gzip {
        #[cfg(feature = "gzip")] {
            use async_compression::tokio::bufread::GzipDecoder;
            use tokio::io::{AsyncReadExt};
            let mut decoder = GzipDecoder::new(tokio::io::BufReader::new(&bytes[..]));
            let mut decompressed = Vec::with_capacity(bytes.len() * 2);
            if let Err(err) = decoder.read_to_end(&mut decompressed).await {
                warn!("HTTP 400 bad request: gzip decompress failed: {}", err);
                return Response::builder()
                    .status(StatusCode::BAD_REQUEST)
                    .body(Full::from(Bytes::from_static(b"invalid gzip body")))
                    .unwrap();
            }
            decompressed
        }
        #[cfg(not(feature = "gzip"))]
        {
            warn!("HTTP request says gzip but binary not built with gzip feature");
            return Response::builder()
                .status(StatusCode::UNSUPPORTED_MEDIA_TYPE)
                .body(Full::from(Bytes::from_static(b"gzip not supported")))
                .unwrap();
        }
    } else {
        bytes.to_vec()
    };
    let decoded = MReq::decode(body_bytes.as_slice());
    let Ok(message) = decoded else {
        return Response::builder()
            .status(StatusCode::BAD_REQUEST)
            .body(Full::from(Bytes::from_static(b"invalid protobuf")))
            .unwrap();
    };

    if let Some(f) = consume {
        f(message).await;
    } else {
        // If no consumer, just warn for traces/metrics; logs currently accepted without consumer
        // Note: Not logging here to keep code generic and quiet by default.
    }

    let resp = MResp::default();
    let mut buf = Vec::with_capacity(64);
    if let Err(_) = prost::Message::encode(&resp, &mut buf) {
        return Response::builder()
            .status(StatusCode::INTERNAL_SERVER_ERROR)
            .body(Full::from(Bytes::from_static(b"encode error")))
            .unwrap();
    }
    Response::builder()
        .status(StatusCode::OK)
        .header(hyper::header::CONTENT_TYPE, "application/x-protobuf")
        .body(Full::from(Bytes::from(buf)))
        .unwrap()
}
